package com.lagou.rpc.balance;

import com.lagou.rpc.netty.RpcClinetChannelHolder;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @ClassName AbstractRpcLoadBanlance
 * @Description
 * @Author playboy
 * @Date 2021/6/9 2:44 下午
 * @Version 1.0
 **/
public abstract class AbstractRpcLoadBanlance implements RpcLoadBanlance, CommandLineRunner, ApplicationContextAware {

	protected AtomicInteger index = new AtomicInteger(0);

	protected String path = "/rpc/servers";

	protected ApplicationContext context;

	protected Map<String, RpcClinetChannelHolder> maps = new ConcurrentHashMap<>();

	protected List<String> servers = new CopyOnWriteArrayList<>();

	protected List<RpcClinetChannelHolder> rpcClinetChannelHolders = new CopyOnWriteArrayList<>();

	CuratorFramework client;

	@Autowired
	ZookeeperProperties zookeeperProperties;

	/**
	 * 负载均衡获取channel，使用长链接
	 *
	 * @return
	 */
	@Override
	public RpcClinetChannelHolder getChannel() {
		RpcClinetChannelHolder rpcClinetChannelHolder = rpcClinetChannelHolders.get(index.getAndAdd(1) % rpcClinetChannelHolders.size());
		System.out.println("当前访问的地址：" + rpcClinetChannelHolder.getInetSocketAddress().getHostString() + ":" + rpcClinetChannelHolder.getInetSocketAddress().getPort());
		return rpcClinetChannelHolder;
	}


	/**
	 * 系统启动完成后执行，连接zookeeper，获取/rpc/下的子节点，并注册变动监听
	 * 根据获取的子节点，创建连接，并且维护到连接池中
	 *
	 * @param args
	 * @throws Exception
	 */
	@Override
	public void run(String... args) throws Exception {
		//在这里进行zookeeper的连接
		ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
		client = CuratorFrameworkFactory.builder()
				.connectString(zookeeperProperties.getAddress())
				.sessionTimeoutMs(50000)
				.connectionTimeoutMs(30000)
				.retryPolicy(retryPolicy)
				.build();
		client.start();

		//获取/rpc下面的子节点
		List<String> nodes = client.getChildren()
				//注册回调函数
				.usingWatcher(new AbstractRpcLoadBanlance.RpcCuratorWatcher())
				.forPath(path);
		handleNodes(nodes);

		start();
	}

	/**
	 * 留给子类实现
	 */
	protected void start() {
	}


	/**
	 * 处理当前的/rpc下面的子节点
	 *
	 * @param nodes
	 */
	private void handleNodes(List<String> nodes) {
		//查找需要新建的连接 即新查询的中包含，但是旧的不包含
		List<String> addNodes = nodes.stream().filter(node -> !servers.contains(node)).collect(Collectors.toList());
		//查找需要关闭的连接 即旧的集合包含，但是新的集合不包含
		List<String> subNodes = servers.stream().filter(node -> !nodes.contains(node)).collect(Collectors.toList());

		addNodes.forEach(node -> {
			//获取当前节点数据
			try {
				String data = node;
				String[] split = data.split(":");
				//根据连接地址创建netty长连接
				RpcClinetChannelHolder rpcClinetChannelHolder = context.getBean(RpcClinetChannelHolder.class);
				InetSocketAddress inetSocketAddress = new InetSocketAddress(split[0], Integer.valueOf(split[1]));
				rpcClinetChannelHolder.setInetSocketAddress(inetSocketAddress);
				rpcClinetChannelHolder.init();
				maps.put(node, rpcClinetChannelHolder);
				rpcClinetChannelHolders.add(rpcClinetChannelHolder);
				servers.add(node);
				System.out.println("建立连接：" + data);
				addFinishHandler(node, rpcClinetChannelHolder);
			} catch (Exception e) {
				e.printStackTrace();
			}
		});

		subNodes.forEach(node -> {
			//获取当前节点数据
			try {
				//将连接关闭，并且移除连接池
				RpcClinetChannelHolder rpcClinetChannelHolder = maps.get(node);
				InetSocketAddress inetSocketAddress = rpcClinetChannelHolder.getInetSocketAddress();
				System.out.println("移除连接：" + inetSocketAddress.getHostString() + ":" + inetSocketAddress.getPort());
				rpcClinetChannelHolder.close();
				maps.remove(node, rpcClinetChannelHolder);
				rpcClinetChannelHolders.remove(rpcClinetChannelHolder);
				servers.remove(node);
				subFinishHandler(node, rpcClinetChannelHolder);
			} catch (Exception e) {
				e.printStackTrace();
			}
		});
	}

	protected void subFinishHandler(String node, RpcClinetChannelHolder rpcClinetChannelHolder) {
	}

	protected void addFinishHandler(String node, RpcClinetChannelHolder rpcClinetChannelHolder) {
	}

	/**
	 * 监听器
	 */
	private class RpcCuratorWatcher implements CuratorWatcher {

		@Override
		public void process(WatchedEvent event) throws Exception {
			//只监控节点的子节点变化事件
			if (event.getType().equals(Watcher.Event.EventType.NodeChildrenChanged)) {
				List<String> nodes = client.getChildren()
						.usingWatcher(this)//重新注册回调函数
						.forPath(path);
				handleNodes(nodes);
			}
		}
	}

	@Override
	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
		this.context = applicationContext;
	}

	/**
	 * 不需要实现
	 *
	 * @param rpcClinetChannelHolder
	 * @param time
	 */
	@Override
	public void setResponseTime(RpcClinetChannelHolder rpcClinetChannelHolder, long time) {

	}

}
